package cn.edu.flink.scala.tutorial.transformer

import cn.edu.flink.scala.tutorial.source.Event
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object 简单聚合算子 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val events = List(
      Event("Mary", "home/", 10000L),
      Event("Mary", "cart/", 20000L),
      Event("Mary", "fav/", 30000L))

    val eventsDS = env.fromCollection(events)

    val keyedDS = eventsDS.keyBy(_.user)

    events.foreach(println)
    println("sum")
    keyedDS.sum("timestamp").executeAndCollect().foreach(println)
    println("max")
    keyedDS.max("timestamp").executeAndCollect().foreach(println)
    println("maxBy")
    keyedDS.maxBy("timestamp").executeAndCollect().foreach(println)
    println("min")
    keyedDS.min("timestamp").executeAndCollect().foreach(println)
    println("minBy")
    keyedDS.minBy("timestamp").executeAndCollect().foreach(println)

  }
}
